package com.bckj.fastboot.iot.datcol.mqtt.core;

import com.bckj.fastboot.core.lang.exception.BusinessException;
import com.bckj.fastboot.iot.common.dto.msg.IotProtocolDataMessage;
import com.bckj.fastboot.iot.common.mq.producer.IotProtocolDataMessageProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Service;

/**
 * MqttCallbackImpl
 *
 * @author wfc
 **/
@Slf4j
@Service
@RequiredArgsConstructor
public class MqttCallbackImpl implements MqttCallback {

    private final IotProtocolDataMessageProducer iotProtocolDataMessageProducer;

    /**
     * MQTT 回调：收到消息时触发
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String payload = new String(message.getPayload());
        log.info("Received message from topic [{}]: {}", topic, payload);
        String[] arr = parseTopic(topic);
        String devModel = arr[0];
        String devCode = arr[1];
        iotProtocolDataMessageProducer.send(IotProtocolDataMessage.ofMqtt(
                devModel, devCode, payload
        ));
    }

    String[] parseTopic(String topic){
        String[] topicArray=topic.split("/");
        if(topic.startsWith("status/handle")){
            if(topicArray.length>=4) {
                return new String[]{ topicArray[2].trim(), topicArray[3].trim()};
            }
        }
        throw new BusinessException("!Topic解析失败 " + topic);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // 消息发送完成回调（可选处理）
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.warn("MQTT Connection lost!", cause);
        // 可以在这里实现断线重连逻辑
    }
}
